package net.bwie.realtime.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dwd.log.utils.AreaInfoUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import utils.DateTimeUtil;
import utils.DorisUtil;
import utils.KafkaUtil;


public class ETLProcess {

    public static void main(String[] args) throws Exception {
        // 1. 创建Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);

        // 2. 从Kafka获取数据源
        DataStream<String> logStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");
//        logStream.print("logStream");

        DataStream<String> stringDataStream = processJson(logStream);
        stringDataStream.print();
        // 5. 输出到Kafka
        KafkaUtil.producerKafka(stringDataStream, "dwd_car_monitor");
        //  数据输出到Doris
        DorisUtil.saveToDoris(stringDataStream,"car_realtime_report","dwd_traffic_etl_log");

        // 6. 执行作业
        env.execute("ETLProcess");

}

    private static DataStream<String> processJson(DataStream<String> stream) {
        //转为实体类，并对时间戳进行毫秒级的转化
        SingleOutputStreamOperator<String> mapStream = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                jsonObject.put("actionTime", DateTimeUtil.convertLongToString(jsonObject.getLong("actionTime") * 1000, DateTimeUtil.DATE_TIME_FORMAT));
                return jsonObject.toJSONString();
            }
        });

        // 3. 调用工具类预加载区域信息
        AreaInfoUtil.preloadAreaInfo();

        // 4. 处理JSON数据（调用工具类的方法）
        DataStream<String> trafficStream = AreaInfoUtil.processJson(mapStream);



        return trafficStream;

    }
 }
